package org.cliff.message.platform.core.aop;

import java.lang.annotation.Annotation;
import java.util.Date;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import cn.hutool.core.exceptions.ExceptionUtil;
import org.cliff.message.platform.core.annotation.MessageConsumer;
import org.cliff.message.platform.core.businessinfo.BusinessInfo;
import org.cliff.message.platform.core.businessinfo.IBusinessInfoRetriever;
import org.cliff.message.platform.core.domain.*;
import org.cliff.message.platform.core.filter.RepeatedlyReadRequestWrapper;
import org.cliff.message.platform.core.mapper.ReceivedMessageDetailMapper;
import org.cliff.message.platform.core.mapper.ReceivedMessageResendDetailMapper;
import org.cliff.message.platform.core.response.CommonResponseBuilder;
import org.cliff.message.platform.core.service.InterfaceConfigMaintainer;
import org.cliff.message.platform.core.service.ReceivedMessageResendService;
import org.cliff.message.platform.core.service.ReceivedMessageService;
import org.cliff.message.platform.core.util.JsonUtil;
import org.cliff.message.platform.core.util.MessagePlatformThreadLocals;
import org.cliff.message.platform.core.util.SpringContextHolder;
import com.xfvape.uid.UidGenerator;
import org.apache.commons.lang.StringUtils;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.*;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import lombok.extern.slf4j.Slf4j;

import static org.cliff.message.platform.core.domain.MPRequestHeaderConstants.AUTO_RETRY_COUNT;
import static org.cliff.message.platform.core.domain.ProcessStatusConstants.RUNNING;
import static org.cliff.message.platform.core.util.MessagePlatformThreadLocals.*;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;

@Slf4j
@Aspect
@Component
@ConditionalOnProperty(value = "messagePlatform.enabled", matchIfMissing = true)
@Order(1)
public class MessageConsumeAopConfig {

    @Autowired
    private CommonResponseBuilder commonResponseBuilder;

    @Autowired
    private InterfaceConfigMaintainer interfaceConfigMaintainer;

    @Autowired
    private UidGenerator uidGenerator;

    @Autowired
    private ReceivedMessageService receivedMessageService;

    @Autowired
    private ReceivedMessageResendService receivedMessageResendService;

    @Autowired
    private ReceivedMessageDetailMapper messageDetailMapper;

    @Autowired
    private ReceivedMessageResendDetailMapper resendDetailMapper;

    /**
     * 定义切点Pointcut 第一个*号：表示返回类型， *号表示所有的类型 第二个*号：表示类名，*号表示所有的类 第三个*号：表示方法名，*号表示所有的方法 后面括弧里面表示方法的参数，两个句点表示任何参数
     */
    @Pointcut("@annotation(org.cliff.message.platform.core.annotation.MessageConsumer)")
    public void messageConsumeExecution() {}

    /**
     * OMS重消费消息header name
     */
    public static final String OMS_RECONSUME_HEADER = "oms_reconsume";


    @Around(value = "messageConsumeExecution() && @annotation(messageConsumer)")
    public Object doAround(ProceedingJoinPoint joinPoint, MessageConsumer messageConsumer) throws Throwable {
        // 由于消息平台组件的ThreadLocal 会被业务代码赋值，但不会进到本aop最终clean处理
        // 故在入口处进行一次clean，保障ThreadLocal不受污染
        MessagePlatformThreadLocals.clean();
        ReceivedMessage receivedMessage = null;
        try {
            receivedMessage = loadReceivedMessage(joinPoint, messageConsumer);
        } catch (Exception e) {
            log.error("Retrieve message failed.", e);
            return joinPoint.proceed(joinPoint.getArgs());
        }
        // 没有幂等性控制
        if (!idempotentControl(joinPoint, messageConsumer)) {
            return joinPoint.proceed(joinPoint.getArgs());
        }

        log.warn("Message {} is ignored for idempotent control.", receivedMessage.getId());
        return commonResponseBuilder.buildIdempotentControlResponse(receivedMessage);
    }

    /**
     * 创建/加载接收消息记录
     *
     * @param joinPoint 连接点
     */
    public ReceivedMessage loadReceivedMessage(JoinPoint joinPoint, MessageConsumer messageConsumer) {
        HttpServletRequest request =
                ((ServletRequestAttributes)RequestContextHolder.getRequestAttributes()).getRequest();
        String resendMessageId = request.getHeader(MPRequestHeaderConstants.RESEND_MESSAGE_ID);

        ReceivedMessage receivedMessage;
        // 重发长江，加载之前拷贝/原始消息记录
        if (StringUtils.isNotEmpty(resendMessageId)) {
            receivedMessage = receivedMessageService.getById(Long.parseLong(resendMessageId));
        }
        // 非重发场景新建记录
        else {
            receivedMessage = buildReceivedMessageDetail(joinPoint, messageConsumer, request);
            receivedMessageService.save(receivedMessage);
        }
        RECEIVED_MESSAGE.set(receivedMessage);
        REQUEST_PROCESS_START_TIME.set(System.currentTimeMillis());
        // 默认所有请求处理都成功
        PROCESS_STATUS.set(ProcessStatusConstants.SUCCEED);
        return receivedMessage;
    }

    /**
     * 幂等性控制
     * @param joinPoint
     * @return
     */
    public boolean idempotentControl(ProceedingJoinPoint joinPoint, MessageConsumer messageConsumer) {
        try {
            if (!messageConsumer.idempotent()) {
                return false;
            }
            // 做幂等性控制但没有获取到businessId，不做幂等性控制
            ReceivedMessage receivedMessage = RECEIVED_MESSAGE.get();
            if (receivedMessage.getBusinessId() == null) {
                log.warn("消息{}设置成要做幂等性控制但没有获取到businessId，跳过幂等性控制", receivedMessage.getId());
                return false;
            }
            // 相同的业务ID是否存在正在处理中的消息

            HttpServletResponse response =
                    ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();
            response.setStatus(HttpStatus.ALREADY_REPORTED.value());
            DATA_STATUS.set(ProcessStatusConstants.WARNING);
            return true;
        } catch (Exception e) {
            log.error("Idempotent control failed.", e);
            return false;
        }
    }

    /**
     * 正常返回处理
     */
    @AfterReturning(pointcut = "messageConsumeExecution()", returning = "returnValue")
    public void doAfterReturning(JoinPoint joinPoint, Object returnValue) {
        try {
            String response = JsonUtil.toJSONString(returnValue);
            receiveMessagePostProcess(response);
        }
        // 捕获所有异常，避免消息平台处理
        catch (Exception e) {
            log.error("Receive message post process failed.", e);
        } finally {
            MessagePlatformThreadLocals.clean();
        }
    }

    /**
     * 抛出异常终止执行处理
     */
    @AfterThrowing(pointcut = "messageConsumeExecution()", throwing = "exception")
    public void doAfterThrowing(JoinPoint joinPoint, Throwable exception) {
        try {
            PROCESS_STATUS.set(ProcessStatusConstants.ERROR);
            receiveMessagePostProcess(ExceptionUtil.stacktraceToString(exception));
        }
        // 捕获所有异常，避免消息平台处理
        catch (Exception e) {
            log.error("Receive message post process failed.", e);
        } finally {
            MessagePlatformThreadLocals.clean();
        }
    }

    /**
     * 接收消息后置处理
     *
     * @param response
     */
    private void receiveMessagePostProcess(String response) {
        // 做幂等性控制但没有获取到businessId，不做幂等性控制
        ReceivedMessage receivedMessage = RECEIVED_MESSAGE.get();
        ReceivedMessageDetail receivedMessageDetail = new ReceivedMessageDetail();
        receivedMessageDetail.setMessageId(receivedMessage.getId());
        receivedMessageDetail.setResponseDetail(response);
        receivedMessageDetail.setProcessStatus(PROCESS_STATUS.get());
        receivedMessageDetail.setDataStatus(DATA_STATUS.get());
        receivedMessageDetail.setProcessTime(System.currentTimeMillis() - REQUEST_PROCESS_START_TIME.get());
        messageDetailMapper.insert(receivedMessageDetail);
        boolean needResend = false;
        // 请求出错且接口需要重试
        if (ProcessStatusConstants.ERROR.equals(receivedMessageDetail.getProcessStatus())) {
            InterfaceConfig interfaceConfig =
                    interfaceConfigMaintainer.getInterfaceConfigByFunctionCode(receivedMessage.getFunctionCode());
            Integer autoRetrySetting = interfaceConfig.getAutoRetryTimes();
            String retryCountStr = ((ServletRequestAttributes)RequestContextHolder.getRequestAttributes()).getRequest()
                    .getHeader(AUTO_RETRY_COUNT);
            int retryCount = StringUtils.isEmpty(retryCountStr) ? 0 : Integer.parseInt(retryCountStr);
            // 接口存在重试配置，且自动重试次数小于当前次数
            if (autoRetrySetting != null && autoRetrySetting > retryCount) {
                HttpServletRequest request =
                        ((ServletRequestAttributes)RequestContextHolder.getRequestAttributes()).getRequest();
                // 重发场景失败在发送端插入重试记录
                if (StringUtils.isEmpty(request.getHeader(MPRequestHeaderConstants.RESEND_MESSAGE_ID))) {
                    receivedMessageResendService.autoResendMessage(receivedMessage, interfaceConfig, retryCount);
                }
                needResend = true;
            }
        }

        // 不重试才更新消息接收主表，否则表明当前消息没有处理完成，不更新消息主表
        if (!needResend) {
            receivedMessage.setProcessStatus(PROCESS_STATUS.get());
            receivedMessage.setDataStatus(DATA_STATUS.get());
            receivedMessage.setUpdateTime(new Date());
            receivedMessageService.updateById(receivedMessage);
        }
    }

    private ReceivedMessage buildReceivedMessageDetail(JoinPoint joinPoint, MessageConsumer messageConsumer,
                                                       HttpServletRequest request) {
        ReceivedMessage receivedMessage = new ReceivedMessage();
        BusinessInfo businessInfo = retrieveBusinessInfo(messageConsumer, joinPoint, request);
        if (businessInfo != null) {
            BeanUtils.copyProperties(businessInfo, receivedMessage);
        }

        receivedMessage.setId(uidGenerator.getUID());
        String requestUri = request.getRequestURI();
        // Function Code加载
        if (StringUtils.isNotEmpty(messageConsumer.functionCode())) {
            receivedMessage.setFunctionCode(messageConsumer.functionCode());
        } else {
            InterfaceConfig interfaceConfig = interfaceConfigMaintainer.getInterfaceConfigByRequestUri(requestUri);
            receivedMessage.setFunctionCode(interfaceConfig.getFunctionCode());
        }
        receivedMessage.setRequestUri(requestUri);
        receivedMessage.setContentType(request.getContentType());

        String body = ((RepeatedlyReadRequestWrapper)request).getBody();
        if (APPLICATION_JSON_VALUE.equalsIgnoreCase(request.getContentType())) {
            receivedMessage.setJsonBody(body);
        } else {
            receivedMessage.setRequestBody(body);
        }
        receivedMessage.setProcessStatus(RUNNING);

        return receivedMessage;
    }

    /**
     * 获取业务信息
     *
     * @param messageConsumer
     * @param joinPoint
     * @param request
     * @return
     */
    private BusinessInfo retrieveBusinessInfo(MessageConsumer messageConsumer, JoinPoint joinPoint,
                                              HttpServletRequest request) {
        if (StringUtils.isNotEmpty(messageConsumer.businessInfoRetriever())) {
            try {
                IBusinessInfoRetriever businessInfoRetriever =
                        SpringContextHolder.getBean(messageConsumer.businessInfoRetriever());
                return businessInfoRetriever.retrieveBusinessInfo(getRequestBody(joinPoint), request);
            } catch (Exception e) {
                log.error("Retrieve business info failed.", e);
            }
        }
        return null;
    }

    /**
     * 获取用@RequestBody 注解声明的请求体
     *
     * @param joinPoint
     * @return
     */
    private Object getRequestBody(JoinPoint joinPoint) {
        Object[] args = joinPoint.getArgs();
        MethodSignature methodSignature = (MethodSignature)joinPoint.getSignature();
        Annotation[][] parameterAnnotations  = methodSignature.getMethod().getParameterAnnotations();
        if (parameterAnnotations != null && parameterAnnotations.length > 0) {
            for (int i = 0; i < parameterAnnotations.length; i++) {
                for (int j = 0; j < parameterAnnotations[i].length; j++) {
                    if (parameterAnnotations[i][j].annotationType() == RequestBody.class && args.length > i) {
                        return args[i];
                    }
                }
            }
        }
        return null;
    }
}
